/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.scheduler;

import java.io.IOException;
import java.util.LinkedList;
import java.util.concurrent.ConcurrentHashMap;

import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.exchange.ConditionFlag;
import cgl.granules.exchange.DeploymentMessage;
import cgl.granules.exchange.ExchangeFactory;
import cgl.granules.exchange.ExecutionStatus;
import cgl.granules.exchange.ResourceDiscovery;
import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventProducer;

/**
 * This class keeps track of various diagnostic-related elements.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class Diagnostics {

    private ConcurrentHashMap<String, TaskLifecycleMetrics> taskInfo;
    private ConcurrentHashMap<String, DeploymentMessage> deploymentRequests;
    private LinkedList<String> completionOrder;
    private int maxNumberOfCompletions = 300000;
    private ExchangeFactory exchangeFactory;
    private StreamingService streamingService;
    private EventProducer producer;
    private String deploymentStream;
    private int numOfThreads;
    private ResourceMetrics resourceMetrics;
    private boolean debug = false;
    private String moduleName = "Diagnostics: ";


    public Diagnostics(String deploymentStream, int numOfThreads)
        throws CommunicationsException {
        this.deploymentStream = deploymentStream;
        this.numOfThreads = numOfThreads;
        resourceMetrics = new ResourceMetrics(numOfThreads, deploymentStream);

        taskInfo = new ConcurrentHashMap<String, TaskLifecycleMetrics>();
        deploymentRequests = new ConcurrentHashMap<String, DeploymentMessage>();
        completionOrder = new LinkedList<String>();
        exchangeFactory = ExchangeFactory.getInstance();
        streamingService = StreamingService.getInstance();
        producer = streamingService.createProducer();
        publishToStream("Test", "Test".getBytes());
    }


    /**
     * Retrieve the resource metrics
     * 
     * @return the resourceMetrics
     */
    public ResourceMetrics getResourceMetrics() {
        return resourceMetrics;
    }


    /**
     * Add the deployment request associated with the task. This method is
     * invoked BEFORE the task is deployed, so if the deployment fails the
     * corresponding method to remove this request will be invoked.
     * 
     * @param taskIdentifier
     *            The id of the task
     * @param deploymentRequest
     *            The deployment request for the task in question.
     */
    public void addDeploymentRequest(String taskIdentifier,
                                     DeploymentMessage deploymentRequest) {
        if (!deploymentRequests.containsKey(taskIdentifier)) {
            deploymentRequests.put(taskIdentifier, deploymentRequest);
        }
    }


    /**
     * Remove the deployment request associated with a task. This method is
     * invoked if there are problems in deploying the operationa associated with
     * the task.
     * 
     * @param taskIdentifier
     *            The id of the task
     */
    public void removeDeploymentRequest(String taskIdentifier) {
        if (deploymentRequests.containsKey(taskIdentifier)) {
            deploymentRequests.remove(taskIdentifier);
        }
    }


    /**
     * Manage information about the newly created task.
     * 
     * @param taskIdentifier
     *            The taskIdentifier assigned to the task
     * @param metrics
     *            The lifecycle metrics associated with the task in question
     */
    public void createdTask(String taskIdentifier, TaskLifecycleMetrics metrics) {
        taskInfo.put(taskIdentifier, metrics);
        reportTaskExecutionStatus(taskIdentifier);
    }


    /**
     * Process the completion of the task.
     * 
     * @param taskIdentifier
     *            The identifier associated with the task.
     */
    public void completedTask(String taskIdentifier) {
        completionOrder.add(taskIdentifier);

        if (completionOrder.size() > maxNumberOfCompletions) {
            String purgeTaskId = completionOrder.poll();
            taskInfo.remove(purgeTaskId);
            deploymentRequests.remove(purgeTaskId);
        }
        
        reportTaskExecutionStatus(taskIdentifier);
    }


    /**
     * Send a report on the task execution status to the appropriate entity.
     * 
     * @param taskIdentifier
     *            The identifier of the task being reported on.
     */
    private void reportTaskExecutionStatus(String taskIdentifier) {
        try {
            if (debug) {
                System.out.println("Sending task execution status for task: "
                    + taskIdentifier);
            }
            sendTaskExecutionStatus(taskIdentifier);
        } catch (SuppressedException e) {
            reportErrorsThatCannotBePropagated(e.toString());
        }
    }


    /**
     * Retrieve the status associated with a specific task.
     * 
     * @param taskIdentifier
     * @return The identifier associated with the task.
     */
    public TaskLifecycleMetrics getStatusOfTask(String taskIdentifier) {
        TaskLifecycleMetrics metrics = taskInfo.get(taskIdentifier);
        return metrics;
    }


    public void averageRateOfTaskArrivals() {

    }


    public void getNumberOfCompletedTasks() {

    }


    public void processResourceDiscoveryRequest(
                                                ResourceDiscovery resourceDiscovery)
        throws SuppressedException {
        String stage = "Processing ResourceDiscovery Exchange";
        String problem = null;
        String respondTo = resourceDiscovery.getRespondTo();
        int conditionFlag = ConditionFlag.NOT_SET;

        if (resourceDiscovery.isResponse()) {
            problem = "Incorrectly received a RESPONSE.";
        }

        if (problem != null) {
            conditionFlag = ConditionFlag.PROBLEMS_ENCOUNTERED;
            sendResourceDiscoveryResponse(conditionFlag, problem, null,
                                          "Annonymous Resource", respondTo);
        } else {
            String description = "Resource Discovery Response";
            resourceMetrics.setNumberOfComputations(taskInfo.size());
            sendResourceDiscoveryResponse(conditionFlag, description,
                                          resourceMetrics, deploymentStream,
                                          respondTo);
        }

    }


    private void sendResourceDiscoveryResponse(int conditionFlag,
                                               String description,
                                               ResourceMetrics metricsToSend,
                                               String respondTo, String sendTo)
        throws SuppressedException {
        ResourceDiscovery resourceDiscoveryResponse =
            exchangeFactory.createResourceDiscoveryResponse(
                                                            conditionFlag,
                                                            "Resource Discovery Response",
                                                            metricsToSend,
                                                            respondTo);

        String stage =
            "Sending an ResourceDiscovery response to [" + respondTo + "]";
        String problem = "";

        try {
            byte[] responseBytes = resourceDiscoveryResponse.getBytes();
            publishToStream(sendTo, responseBytes);
        } catch (IOException e) {
            problem = " there was an IOException " + e.getMessage();
            throwSupressedException(stage, problem);
        } catch (MarshallingException e) {
            problem = " there was a MarshallingException " + e.getMessage();
            throwSupressedException(stage, problem);
        } catch (CommunicationsException e) {
            problem = " there was a CommunicationsException " + e.getMessage();
            throwSupressedException(stage, problem);
        }
    }


    public void processExecutionStatusRequest(ExecutionStatus executionStatus)
        throws SuppressedException {
        String stage = "Processing ExecutionStatus Exchange";
        String problem = null;
        String jobIdentifier = executionStatus.getJobIdentifier();
        String taskIdentifier = executionStatus.getTaskIdentifier();
        String respondTo = executionStatus.getRespondTo();
        int conditionFlag = ConditionFlag.NOT_SET;
        TaskLifecycleMetrics taskLifecycleMetrics =
            getStatusOfTask(taskIdentifier);

        if (executionStatus.isResponse()) {
            problem = "Incorrectly received a RESPONSE.";
        }

        if (taskLifecycleMetrics == null) {
            problem =
                "Unknown taskId specified in the request. Perhaps, "
                    + "information about this task has been reclaimed.";
        }

        if (problem != null) {
            conditionFlag = ConditionFlag.PROBLEMS_ENCOUNTERED;
            sendTaskExecutionStatus(jobIdentifier, taskIdentifier,
                                    conditionFlag, problem,
                                    taskLifecycleMetrics, respondTo);
        }

        sendTaskExecutionStatus(jobIdentifier, taskIdentifier, respondTo,
                                taskLifecycleMetrics);
    }


    private void sendTaskExecutionStatus(String taskIdentifier)
        throws SuppressedException {
        DeploymentMessage deploymentRequest =
            deploymentRequests.get(taskIdentifier);
        if (deploymentRequest == null) {
            throw new SuppressedException(moduleName + "Unable to locate "
                + "DeploymentRequest associated with task" + taskIdentifier);
        }
        String jobIdentifier = deploymentRequest.getJobIdentifier();
        String respondTo = deploymentRequest.getRespondTo();/* won't be null. */

        TaskLifecycleMetrics taskLifecycleMetrics =
            getStatusOfTask(taskIdentifier);
        sendTaskExecutionStatus(jobIdentifier, taskIdentifier, respondTo,
                                taskLifecycleMetrics);
    }


    /**
     * Create a ExecutionStatus Response, based on the specified parameters and
     * send it along to requester.
     */
    private void sendTaskExecutionStatus(
                                         String jobIdentifier,
                                         String taskIdentifier,
                                         String respondTo,
                                         TaskLifecycleMetrics taskLifecycleMetrics)
        throws SuppressedException {
        int status = taskLifecycleMetrics.getExecutionStatus();
        int conditionFlag = getConditionValue(status);

        String description = taskLifecycleMetrics.toString();
        sendTaskExecutionStatus(jobIdentifier, taskIdentifier, conditionFlag,
                                description, taskLifecycleMetrics, respondTo);
    }


    /** Create the ExecutionStatus response, and send it along to the requester. */
    private void sendTaskExecutionStatus(
                                         String jobIdentifier,
                                         String taskIdentifier,
                                         int conditionFlag,
                                         String description,
                                         TaskLifecycleMetrics taskLifecycleMetrics,
                                         String respondTo)
        throws SuppressedException {

        ExecutionStatus executionStatusResponse =
            exchangeFactory.createExecutionStatusResponse(jobIdentifier,
                                                          taskIdentifier,
                                                          conditionFlag,
                                                          description,
                                                          taskLifecycleMetrics,
                                                          deploymentStream);
        String stage = "Sending a TaskExecutionStatus to [" + respondTo + "]";
        String problem = "";

        try {
            byte[] responseBytes = executionStatusResponse.getBytes();
            publishToStream(respondTo, responseBytes);
        } catch (IOException e) {
            problem = " there was an IOException " + e.getMessage();
            throwSupressedException(stage, problem);
        } catch (MarshallingException e) {
            problem = " there was a MarshallingException " + e.getMessage();
            throwSupressedException(stage, problem);
        } catch (CommunicationsException e) {
            problem = " there was a CommunicationsException " + e.getMessage();
            throwSupressedException(stage, problem);
        }
    }


    /**
     * Set the condition variable based on the values contained in the
     * TaskLifeCyleMetrics, which provides a much finer-grained description of
     * the status of the execution.
     * 
     * @param status
     * @return The mapped condition value
     */
    private int getConditionValue(int status) {
        int conditionFlag = ConditionFlag.NOT_SET;
        if (status == TaskLifecycleMetrics.EXECUTING
            || status == TaskLifecycleMetrics.INITIALIZED
            || status == TaskLifecycleMetrics.INITIALIZING
            || status == TaskLifecycleMetrics.QUEUED_FOR_EXECUTION
            || status == TaskLifecycleMetrics.WAITING_FOR_DATA
            || status == TaskLifecycleMetrics.READY) {
            conditionFlag = ConditionFlag.EXECUTING;
        }

        if (status == TaskLifecycleMetrics.COMPLETE) {
            conditionFlag = ConditionFlag.SUCCESS;
        }

        if (status == TaskLifecycleMetrics.ABNORMAL_TERMINATION
            || status == TaskLifecycleMetrics.ABORTED) {
            conditionFlag = ConditionFlag.FAILURE;
        }
        return conditionFlag;
    }


    /**
     * Utility method that allows a producer to publish a set of bytes onto the
     * specified stream.
     * 
     * @param publishStream
     *            Synopsis of the stream to publish to
     * @param transferBytes
     *            the payload for the stream fragments
     * @throws CommunicationsException
     *             When communication problems are encountered.
     */
    private void publishToStream(String publishStream, byte[] transferBytes)
        throws CommunicationsException {
        // System.out.println(moduleName + "Publishing to Stream " +
        // publishStream);
        NBEvent nbEvent;
        try {
            if (transferBytes == null) {
                System.out.println(moduleName + "Trying to send NULL bytes");
                return;
            }
            nbEvent =
                producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
                                       publishStream, transferBytes);
            producer.publishEvent(nbEvent);
        } catch (ServiceException e) {
            e.printStackTrace();
            throw new CommunicationsException(e.toString());
        }
    }


    private void reportErrorsThatCannotBePropagated(String errorReport) {
        String completeError =
            moduleName + "This is an error that CANNOT be propagated."
                + errorReport + " No further processing will be done";
        System.out.println(completeError);
    }


    private void throwSupressedException(String stage, String problem)
        throws SuppressedException {
        String errorReport =
            "Problems encountered when (" + stage + ") because :-> " + problem;
        throw new SuppressedException(errorReport);
    }
}
